package com.atguigu.gmall.sink;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.bean.TableProcess;
import com.atguigu.gmall.util.JdbcUtil;
import com.atguigu.gmall.util.RedisUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/30 10:29
 * 通过Phoenix写入hbase
 * <p>
 * Phoenix支持sql, 所以连接Phoenix可以使用标准的jdbc
 * <p>
 * 1. flink提供了一个jdbcSink, 可以对JdbcSink做封装, 来实现PhoenixSink
 * <p>
 * ----
 * <p>
 * 1. 建表
 * <p>
 * 2. 写数据
 */
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {
    
    private Connection conn;
    private ValueState<String> tableCreateState;
    private Jedis client;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        // 先加载驱动, 很多情况不是必须. 大部分常用的数据库会根据url自动选择合适的driver.   Phoenix 驱动有些时候需要手动加载一下
        // 获取连接.
        conn = JdbcUtil.getPhoenixConnection();
        
        tableCreateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("tableCreateState", String.class));
        
        client = RedisUtil.getRedisClient();
        client.select(1);
    }
    
    //使用标准的jdbc来完成
    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        // 1. 检测表, 如果表不存在就需要在Phoenix中新建表
        checkTable(value);
        // 2. 再把数据写入到phoenix中
        writeToPhoenix(value);
        
        // 3. 更新redis缓存  (读取维度数据缓存优化的时候, 添加这个代码)
        // 对缓存中已经存在的, 并且这次又更新的维度, 去更新缓冲中的维度. 新增的维度千万不要去写入到缓存
        // 粗暴: 直接把缓存删除
        updateCache(value);
    }
    
    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        // {"id": 1} => {"ID": 1}
        
        // 把json数据的key全部大写
        JSONObject data = new JSONObject();
        for (Map.Entry<String, Object> entry : value.f0.entrySet()) {
            data.put(entry.getKey().toUpperCase(), entry.getValue());
        }
    
        // 更新redis缓存
        // 这次是update, 并且redis中还存在
        String operateType = value.f1.getOperate_type();
        String key = value.f1.getSink_table().toUpperCase() + ":" + data.get("ID");
        
        String dim = client.get(key);
        if ("update".equals(operateType) && dim != null) {
            // 更新
            client.setex(key, 24 * 60 * 60, data.toJSONString());
        }
    }
    
    private void writeToPhoenix(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        // upsert  into user(id, name, age) values(?,?,?)
        StringBuilder insertSql = new StringBuilder();
        // 拼接SQL语句
        insertSql
            .append("upsert into ")
            .append(tp.getSink_table())
            .append("(")
            // 字段名
            .append(tp.getSink_columns())
            .append(")values(")
            // 拼接 ?
            .append(tp.getSink_columns().replaceAll("[^,]+", "?"))
            .append(")");
        
        PreparedStatement ps = conn.prepareStatement(insertSql.toString());
        // 给ps中的占位符赋值
        String[] columnNames = tp.getSink_columns().split(",");
        for (int i = 0; i < columnNames.length; i++) {
            Object v = data.get(columnNames[i]);
            ps.setString(i + 1, v == null ? "" : v.toString());  // null -> "null"
        }
        
        ps.execute();
        conn.commit();
        ps.close();
        
    }
    
    private void checkTable(Tuple2<JSONObject, TableProcess> value) throws SQLException, IOException {
        if (tableCreateState.value() == null) {
            // 执行sql语句   create table if not exists user(id varchar, age varchar )
            TableProcess tp = value.f1;
            StringBuilder createSql = new StringBuilder();
            // 拼接sql语句
            createSql
                .append("create table if not exists ")
                .append(tp.getSink_table())
                .append("(")
                .append(tp.getSink_columns().replaceAll(",", " varchar,"))
                .append(" varchar, constraint pk primary key(")
                .append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                .append("))")
                .append(tp.getSink_extend() == null ? "" : tp.getSink_extend());
            
            System.out.println(createSql.toString());
            PreparedStatement ps = conn.prepareStatement(createSql.toString());
            ps.execute();
            conn.commit();
            ps.close();
            
            tableCreateState.update(tp.getSink_table());
        }
        
    }
}
